package net.bwie.vehicle.dws.function1;

import net.bwie.vehicle.dws.bean1.VehicleData;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

// HeatmapCounter now extends ProcessWindowFunction instead of ProcessAllWindowFunction
public   class HeatmapCounterFunction extends ProcessWindowFunction<VehicleData, Map<String, Integer>, String, TimeWindow> {
/**
 * 处理给定键的车辆数据，生成热图数据
 * 此方法的目的是遍历特定键的车辆数据集，为每个唯一的键计算出现次数，
 * 最终生成一个热图数据结构，其中每个键关联的值表示该键在车辆数据集中出现的次数
 *
 * @param key       车辆数据的唯一键，用于标识和分组车辆数据
 * @param context   上下文对象，可能用于提供有关当前处理环境的信息（未在当前代码中使用，但可能在更广泛的应用程序中有用途）
 * @param elements  车辆数据的集合，表示一系列车辆数据记录，用于生成热图
 * @param out       一个收集器，用于输出处理后的热图数据结构
 */
public void process(String key, Context context, Iterable<VehicleData> elements, Collector<Map<String, Integer>> out) {
    // 初始化热图数据结构，用于存储每个键及其出现次数的映射
    Map<String, Integer> heatmap = new HashMap<>();
    // 遍历车辆数据集，为每个数据项更新热图
    for (VehicleData data : elements) {
        // 更新热图中当前键的计数，如果键不存在，则初始化为0后加1
        heatmap.put(key, heatmap.getOrDefault(key, 0) + 1);
    }
    // 使用收集器输出热图数据结构
    out.collect(heatmap);
}

}